package com.qiaofang.housewarehouse.dao.hbase;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CoprocessorEnvironment;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.transport.client.PreBuiltTransportClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetAddress;
import java.util.*;

public class HabseESRentalObServer extends BaseRegionObserver {
    private static final Logger logger = LoggerFactory.getLogger(HabseESRentalObServer.class);

    public static TransportClient client = null;
    private static String clusterName = "es-test-cluster";//设置集群名称
    private static boolean sniff = true;// 是否自动检测变化节点
    private static String pingTimeout = "5s";// 等待来自节点的ping响应的时间。默认为5 s
    private static boolean ignoreClusterName = false;//是否忽略集群名称
    private static String nodesSamplerInterval = "5s";// 对列出和连接的节点进行采样的频率。默认为5 s
    private static String IP = "192.168.1.153";
    private static int PORT = 9300;  //端口

    @Override
    public void start(CoprocessorEnvironment e) throws IOException {
        Settings settings = Settings.builder().put("cluster.name", clusterName).put("client.transport.sniff", sniff)
                .put("client.transport.ignore_cluster_name", ignoreClusterName)
                .put("client.transport.ping_timeout", pingTimeout)
                .put("client.transport.nodes_sampler_interval", nodesSamplerInterval).build();
        client = new PreBuiltTransportClient(settings).addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(IP), Integer.valueOf(PORT)));

    }
    @Override
    public void postPut(ObserverContext<RegionCoprocessorEnvironment> e, Put put, WALEdit edit, Durability durability) throws IOException {
        try {
            String id = new String(put.getRow());
            NavigableMap<byte[], List<Cell>> familyCellMap = put.getFamilyCellMap();
            Map<String, Object> json = new HashMap<>();
            for (Map.Entry<byte[], List<Cell>> entry : familyCellMap.entrySet()) {
                for (Cell cell : entry.getValue()) {
                    String key = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String value = Bytes.toString(CellUtil.cloneValue(cell));
                    json.put(key, value);
                }
            }
            client.prepareIndex("house-demo", "rental",id).setSource(json).execute().actionGet();
            logger.info("observer -- add new doc: " + id + " to type: house-demo ");
        } catch (Exception x){
            logger.error("------------sync hbase data to es error " + x );
        }
    }

    private static void testGetPutData(String rowKey, String columnFamily, String column, String value) {
        Put put = new Put(Bytes.toBytes(rowKey));
        put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value));

        NavigableMap<byte[], List<Cell>> familyMap = put.getFamilyCellMap();
        Map<String, Object> json = new HashMap<>();
        System.out.println(Bytes.toString(put.getRow()));
        for (Map.Entry<byte[], List<Cell>> entry : familyMap.entrySet()) {
            for (Cell cell : entry.getValue()) {
                String key = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value1 = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.println(key);
                System.out.println(value1);
                json.put(key, value1);
            }
        }
        System.out.println(json.toString());
    }
    public static void main(String[] args) {
        testGetPutData("上海", "info", "c1", "hello world");
    }
}
